package com.jerome.election.server.paxos;

import com.jerome.election.server.ContextUtil;
import com.jerome.election.server.MemberStateEnum;
import com.jerome.election.server.PaxosMethodEnum;
import com.jerome.election.storage.ExpiringMap;
import com.jerome.election.sync.SyncClient;
import com.jerome.election.sync.SyncVo;
import com.jerome.election.util.ThreadPoolFactory;

import java.io.IOException;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Proposer and Accepter
 *
 * @author Jerome
 * @version 1.0
 * @date 2023/10/19 14:36
 */
public class ProposeAccepter {
    /**
     * Proposal number
     */
    private int proposalNumber = 0;
    /**
     * Proposal value
     */
    private String proposerName;
    /**
     * Current leader
     */
    private PaxosMessage leader;
    /**
     * State last time
     */
    private long lastTime = System.currentTimeMillis() + 10000;

    private MemberStateEnum stateEnum;

    ThreadPoolExecutor threadPool;

    public void setLastTime(long time) {
        lastTime = time;
    }

    public long getLastTime() {
        return lastTime;
    }

    public ProposeAccepter(String proposerName, MemberStateEnum stateEnum) {
        this.proposerName = proposerName;
        this.stateEnum = stateEnum;
        threadPool = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(1), new ThreadPoolFactory("ProposeAccepter"));
        // Start a thread
        threadPool.submit(new KeepAliveWatchDog());
    }

    /**
     * Proposal, candidate proposal expires after 15 seconds
     */
    private ExpiringMap<Integer, Integer> quasiProposals = new ExpiringMap<>(15000);
    /**
     * Accept, candidate proposal expires after 15 seconds
     */
    private ExpiringMap<Integer, Integer> quasiAccepts = new ExpiringMap<>(15000);

    public PaxosMessage getLeader() {
        return leader;
    }

    /**
     * Get and increment votes for the proposal
     *
     * @return proposalNumber
     */
    public synchronized void setQuasiProposalsVotes(int proposalNumber) throws IOException {
        if (quasiAccepts.get(proposalNumber) == null) {
            Integer votes = setProposalsVotes(quasiProposals, proposalNumber);
            if (votes >= ContextUtil.getMajorNum()) {
                // Enter the accept phase
                System.out.println("Proposer: proposal passed");
                SyncVo syncVo = new SyncVo(PaxosMethodEnum.ACCEPT.getCode(), proposalNumber, proposerName);
                send(syncVo);
            }
        }
    }

    /**
     * 发送消息
     *
     * @param syncVo
     */
    private void send(SyncVo syncVo) {
        for (Map.Entry<Integer, SyncClient> entry : ContextUtil.getAcceptors().entrySet()) {
            if (ContextUtil.getNodeMap().get(entry.getKey()) != null) {
                try {
                    entry.getValue().send(syncVo);
                } catch (IOException e) {
                    System.out.println("ProposeAccepter send fail");
                }
            }
        }
    }

    /**
     * Get and increment votes for the accept proposal
     *
     * @return proposalNumber
     */
    public synchronized void setQuasiAcceptsVotes(int proposalNumber) {
        Integer votes = setProposalsVotes(quasiAccepts, proposalNumber);
        if (votes >= ContextUtil.getMajorNum() && !MemberStateEnum.LEADER.equals(stateEnum)) {
            // Accept proposal passed
            System.out.println(String.format("Proposer: accept passed, proposalNumber:%d, proposalValue:%s", proposalNumber, proposerName));
            // Update state
            stateEnum = MemberStateEnum.LEADER;
            lastTime = System.currentTimeMillis();
        }
    }

    /**
     * Get and increment votes for candidate proposals
     *
     * @return proposalNumber
     */
    private synchronized Integer setProposalsVotes(ExpiringMap<Integer, Integer> quasiProposals, int proposalNumber) {
        Integer count = quasiProposals.get(proposalNumber);
        if (count == null) {
            count = 1;
        } else {
            count++;
        }
        quasiProposals.put(proposalNumber, count);
        return count;
    }

    public int getProposalNumber() {
        return proposalNumber;
    }

    public MemberStateEnum getStateEnum() {
        return stateEnum;
    }

    /**
     * Accept prepare proposal and update proposal number
     *
     * @param proposalNumber Proposal number
     * @return boolean true if accepted, false otherwise
     */
    public synchronized boolean prepareProposalNumber(int proposalNumber) {
        if (proposalNumber > this.proposalNumber) {
            this.proposalNumber = proposalNumber;
            lastTime = System.currentTimeMillis();
            if (!MemberStateEnum.ACCEPTER.equals(stateEnum) && !MemberStateEnum.LEADER.equals(stateEnum)) {
                stateEnum = MemberStateEnum.FOLLOWER;
            }
            return true;
        }
        return false;
    }

    /**
     * Accept accept proposal
     *
     * @param proposalNumber Proposal number
     * @return boolean true if accepted, false otherwise
     */
    public synchronized boolean acceptProposalNumber(int proposalNumber) {
        if (proposalNumber >= this.proposalNumber) {
            lastTime = System.currentTimeMillis();
            if (!MemberStateEnum.ACCEPTER.equals(stateEnum)) {
                stateEnum = MemberStateEnum.FOLLOWER;
            }
            return true;
        }
        return false;
    }

    /**
     * Receive the leader's heartbeat
     *
     * @param paxosMessage
     * @author Jerome
     * @date 2023/10/19 17:21
     */
    public void heartbeat(PaxosMessage paxosMessage) {
        if (paxosMessage.getProposalNumber() >= proposalNumber) {
            // Update locally
            setLastTime(System.currentTimeMillis());
            // If the current thread does not need adjustment
            if (!MemberStateEnum.ACCEPTER.equals(stateEnum) && !proposerName.equals(paxosMessage.getProposalValue())) {
                stateEnum = MemberStateEnum.FOLLOWER;
            }
            leader = paxosMessage;
        }
    }

    /**
     * Watchdog for keeping the connection alive
     */
    private class KeepAliveWatchDog implements Runnable {
        Random random = new Random();
        /**
         * Random timeout between 100 - 500 milliseconds
         */
        int sleepTime = random.nextInt(400) + 100;

        @Override
        public void run() {
            while (true) {
                try {
                    if (System.currentTimeMillis() - lastTime > sleepTime) {
                        leader = null;
                        if (!MemberStateEnum.ACCEPTER.equals(stateEnum)) {
                            // Change state
                            stateEnum = MemberStateEnum.CANDIDATE;
                        }
                    }
                    switch (stateEnum) {
                        case LEADER:
                            // Send heartbeat
                            SyncVo leader = new SyncVo(PaxosMethodEnum.LEADER.getCode(), proposalNumber, proposerName);
                            send(leader);
                            break;
                        case ACCEPTER:
                        case FOLLOWER:
                            try {
                                // Sleep for a random number of milliseconds
                                Thread.sleep(sleepTime);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            break;
                        case CANDIDATE:
                            //判断活跃node
                            ContextUtil.getNodeMap().refresh();
                            if (ContextUtil.getNodeMap().keySet().size() >= ContextUtil.getMajorNum()) {
                                // Participate in the election
                                SyncVo prepare = new SyncVo(PaxosMethodEnum.PREPARE.getCode(), proposalNumber + 1, proposerName);
                                send(prepare);
                            }
                            try {
                                // After participating in the election
                                // Sleep for randomNumber/2 milliseconds
                                Thread.sleep(sleepTime / 2);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            break;
                        default:
                            break;
                    }
                    try {
                        Thread.sleep(20);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}



